package main

import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"time"

	"gitee.com/czy233/go-ceph/rados"
	sarama "github.com/Shopify/sarama"
)

var (
	caCert0 = "./gen_crypto/crypto-config/peerOrganizations/org1.demo.com/peers/peer0.org1.demo.com/tls/ca.crt"
	caCert1 = "./gen_crypto/crypto-config/peerOrganizations/org2.demo.com/peers/peer0.org2.demo.com/tls/ca.crt" // client's ca
	caCert2 = "./gen_crypto/crypto-config/peerOrganizations/org3.demo.com/peers/peer0.org3.demo.com/tls/ca.crt"
	caCert3 = "./gen_crypto/crypto-config/peerOrganizations/org4.demo.com/peers/peer0.org4.demo.com/tls/ca.crt"

	serverCert0 = "./gen_crypto/crypto-config/peerOrganizations/org1.demo.com/peers/peer0.org1.demo.com/tls/server.crt" // server's cert
	serverKey0  = "./gen_crypto/crypto-config/peerOrganizations/org1.demo.com/peers/peer0.org1.demo.com/tls/server.key" // server's private key
	serverCert1 = "./gen_crypto/crypto-config/peerOrganizations/org2.demo.com/peers/peer0.org2.demo.com/tls/server.crt" // server's cert
	serverKey1  = "./gen_crypto/crypto-config/peerOrganizations/org2.demo.com/peers/peer0.org2.demo.com/tls/server.key" // server's private key
	serverCert2 = "./gen_crypto/crypto-config/peerOrganizations/org3.demo.com/peers/peer0.org3.demo.com/tls/server.crt" // server's cert
	serverKey2  = "./gen_crypto/crypto-config/peerOrganizations/org3.demo.com/peers/peer0.org3.demo.com/tls/server.key" // server's private key
	serverCert3 = "./gen_crypto/crypto-config/peerOrganizations/org4.demo.com/peers/peer0.org4.demo.com/tls/server.crt" // server's cert
	serverKey3  = "./gen_crypto/crypto-config/peerOrganizations/org4.demo.com/peers/peer0.org4.demo.com/tls/server.key" // server's private key
)

type Server struct {
	ioctx  *rados.IOContext
	conn   *rados.Conn
	time   int
	From   int
	repair chan Repair
	time1  time.Time
}

type Repair struct {
	obj    string
	offset int64
}

type ob struct {
	name    string
	cluster int
}

func test(a chan Repair, cli *http.Client) {
	var c Ceph
	c.CephInit()
	for {
		obj := <-a
		log.Println("cluster", Cluster, ":", "开始恢复", obj.obj, obj.offset)
		c.RepairFunc(obj, cli)
		log.Println("cluster", Cluster, ":", "恢复完成", obj.obj, obj.offset)
	}
}

func (ceph *Ceph) RepairFunc(repair Repair, cli *http.Client) error {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V0_11_0_2

	log.Println("cluster", Cluster, ":", "启动恢复用kafka")

	configc := sarama.NewConfig()
	configc.Consumer.Return.Errors = true
	configc.Version = sarama.V0_11_0_2

	// consumer
	consumer, err := sarama.NewConsumer([]string{KafkaIpAddress}, configc)
	if err != nil {
		log.Println("cluster", Cluster, ":", "consumer_test create consumer error %s\n", err.Error())
		return err
	}

	defer consumer.Close()

	partition_consumer, err := consumer.ConsumePartition("block", 0, repair.offset)
	defer partition_consumer.Close()
	if err != nil {
		log.Println("cluster", Cluster, ":", "恢复时读取区块失败:", repair.offset)
		return errors.New("恢复时读取区块失败")
	}
	msg := <-partition_consumer.Messages()
	bbb := Block2{}
	var bb Block
	err = json.Unmarshal(msg.Value, &bbb)
	bb.BlockId = int(msg.Offset)
	bb.Offset64 = msg.Offset
	bb.Set(bbb, int(msg.Offset))

	q := GenerateDateDistribute(bb)

	var a, b, c ob
	for _, v := range q.Quin {
		if repair.obj == string(v.Trans[0].Hash) {
			a.name = string(v.Trans[1].Hash)
			a.cluster = int(v.Trans[1].From)
			b.name = string(v.Trans[2].Hash)
			b.cluster = int(v.Trans[2].From)
			c.name = "_4" + repair.obj + "+" + a.name + "+" + b.name
			c.cluster = int(v.Storage[0])
			break
		}
		if repair.obj == string(v.Trans[1].Hash) {
			a.name = string(v.Trans[0].Hash)
			a.cluster = int(v.Trans[0].From)
			b.name = string(v.Trans[2].Hash)
			b.cluster = int(v.Trans[2].From)
			c.name = "_4" + a.name + "+" + repair.obj + "+" + b.name
			c.cluster = int(v.Storage[0])
			break
		}
		if repair.obj == string(v.Trans[2].Hash) {
			a.name = string(v.Trans[0].Hash)
			a.cluster = int(v.Trans[0].From)
			b.name = string(v.Trans[1].Hash)
			b.cluster = int(v.Trans[1].From)
			c.name = "_4" + a.name + "+" + b.name + "+" + repair.obj
			c.cluster = int(v.Storage[0])
			break
		}
		if repair.obj == string(v.Trans[0].Hash)+string(v.Trans[1].Hash)+string(v.Trans[2].Hash) {

		}
	}

	data1 := make([]byte, 1024*1024*4)
	data2 := make([]byte, 1024*1024*4)
	data3 := make([]byte, 1024*1024*4)
	data4 := make([]byte, 1024*1024*4)

	data1, err = GetBlockFromTLS(a.name, a.cluster, cli)
	if err != nil || data1 == nil {
		log.Println("cluster", Cluster, ":", "恢复时获取数据失败:", repair, a.name, a.cluster)
		return errors.New("")
	}
	data2, _ = GetBlockFromTLS(b.name, b.cluster, cli)
	if err != nil || data2 == nil {
		log.Println("cluster", Cluster, ":", "恢复时获取数据失败:", repair, b.name, b.cluster)
		return errors.New("")
	}
	data3, _ = GetBlockFromTLS(c.name, c.cluster, cli)
	if err != nil || data3 == nil {
		log.Println("cluster", Cluster, ":", "恢复时获取数据失败:", repair, c.name, c.cluster)
		return errors.New("")
	}
	// s.ioctx.Read(a, data1, 0)
	// s.ioctx.Read(b, data2, 0)
	// s.ioctx.Read(c, data3, 0)

	//TODO: 换成AVX
	for i := 0; i < 1024*1024*4; i++ {
		data4[i] = data1[i] + data2[i] + data3[i]
	}

	err = ceph.writedatapartial(repair.obj, data4)
	if err != nil {
		log.Println("cluster", Cluster, ":", "恢复时写入错误", repair.obj)
		return err
	}

	//TODO: 记得删掉
	data := make([]byte, 1024*1024*4)
	ceph.ioctx.Read(repair.obj, data, 0)
	for i := 0; i < 1024*1024*4; i++ {
		if data4[i] != data[i] {
			log.Println("cluster", Cluster, ":", "恢复时失败", repair.obj)
		}
	}
	log.Println("cluster", Cluster, ":", "恢复成功", repair.obj)
	return nil
}

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	defer r.Body.Close()

	if s.time == 300 {
		s.time = 1
		if s.time1.Sub(time.Now()).Seconds() < 1 {
			log.Println("cluster", Cluster, ":", "拒绝服务")
			return
		}
		s.time1 = time.Now()
	}

	log.Println("cluster", Cluster, ":", "抽查时间", time.Now())

	requesttype := r.FormValue("type")
	if requesttype == "merkleproof" {
		oid := r.FormValue("oid")
		off, err := strconv.ParseUint(r.FormValue("off"), 10, 64)

		if s.time == 10 {
			r := Repair{obj: oid, offset: int64(off)}
			s.repair <- r
		}

		offset := 0

		var data []byte
		data = make([]byte, 4096*1024)
		dlen, err := s.ioctx.Read(oid, data, 0)
		if err != nil {
			log.Println("cluster", Cluster, ":", "错误 被抽查读取 obj:", err)
			return
		}
		if dlen != 4096*1024 {
			log.Println("cluster", Cluster, ":", "错误 数据长度不足 被抽查读取 obj:", err)
			return
		}

		var m MerkleTree
		m.InitConfig([]int{1, 32, 32})
		err = m.GenerateTree(data)
		if err != nil {
			log.Println("cluster", Cluster, ":", "generate mktree error", "oid", oid, "offset", offset)
			return
		}
		proof, err := m.GenerateMerkleProof(int(offset))
		//fmt.Println(proof)
		if err != nil {
			log.Println("cluster", Cluster, ":", "generate mktree proof error", "oid", oid, "offset", offset)
			return
		}
		proof_byte, err := json.Marshal(proof)

		send_len, _ := w.Write(proof_byte)
		log.Println("cluster", Cluster, ":", "生成证明:", oid, ", offset:", offset, proof.Proof[len(proof.Proof)-1][0], "sendlength: ", send_len)

		s.time++
		log.Println("cluster", Cluster, ":", "请求来源", s.From, "累计次数", s.time)
	} else if requesttype == "repair" {
		oid := r.FormValue("oid")

		//TODO:
		if len(oid) > 30 {
			oid = strings.Replace(oid, " ", "+", -1)
		}
		fmt.Println("oid:", oid)

		data := make([]byte, 4096*1024)
		dlen, err := s.ioctx.Read(oid, data, 0)
		if err != nil {
			log.Println("cluster", Cluster, ":", "错误 响应恢复:读取", oid, "err:", err)
			return
		}
		if dlen != 4096*1024 {
			log.Println("cluster", Cluster, ":", "错误 数据长度不足 响应恢复:读取 ", oid, dlen, err)
			return
		}
		send_len, _ := w.Write(data)
		log.Println("cluster", Cluster, ":", "响应恢复:", oid, "sendlength: ", send_len)
	}
}

func tls_server(cluster int, test chan Repair) {
	conn, err := rados.NewConn()
	if err != nil {
		log.Println("cluster", Cluster, ":", "error when invoke a new connection:", err)
		return
	}
	defer conn.Shutdown()

	err = conn.ReadConfigFile("/etc/ceph/ceph.conf")
	if err != nil {
		log.Println("cluster", Cluster, ":", "error when read default config file:", err)
		return
	}

	err = conn.Connect()
	if err != nil {
		log.Println("cluster", Cluster, ":", "error when connect:", err)
		return
	}

	log.Println("cluster", Cluster, ":", "connect ceph cluster ok!")

	ioctx, err := conn.OpenIOContext("testbench")
	if err != nil {
		log.Println("cluster", Cluster, ":", "error when open IO context:", err)
		return
	}

	caCert := ClustercaCert(cluster)
	pool := x509.NewCertPool()
	addTrust(pool, caCert)

	s := Server{ioctx: ioctx, conn: conn, From: cluster, repair: test, time1: time.Now()}

	server := &http.Server{
		Addr:    ":1000" + strconv.Itoa(cluster),
		Handler: &s,
		TLSConfig: &tls.Config{
			ClientCAs:  pool,
			ClientAuth: tls.RequireAndVerifyClientCert,
		},
	}

	serverCert, serverKey := ClusterServer()

	err = server.ListenAndServeTLS(serverCert, serverKey) //添加服务端证书和密钥
	if err != nil {
		log.Println("cluster", Cluster, ":", "ListenAndServeTLS err:", err)
	}

	fmt.Println(pool.Subjects())
}

func ClusterServer() (certFile string, keyFile string) {
	switch Cluster {
	case 0:
		certFile = serverCert0
		keyFile = serverKey0
	case 1:
		certFile = serverCert1
		keyFile = serverKey1
	case 2:
		certFile = serverCert2
		keyFile = serverKey2
	case 3:
		certFile = serverCert3
		keyFile = serverKey3
	}
	return certFile, keyFile

}

func ClustercaCert(cluster int) (certFile string) {
	switch cluster {
	case 0:
		certFile = caCert0
	case 1:
		certFile = caCert1
	case 2:
		certFile = caCert2
	case 3:
		certFile = caCert3
	}
	return certFile
}
